package com.fury.demos.workers;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class ImprovedWorkerDemo {

	private static final int NUMBER_OF_JOBS = 1000;
	private static final int WORKERS = Runtime.getRuntime().availableProcessors();

	private final CyclicBarrier startBarrier;
	private final BlockingQueue<Job> queue;
	private CyclicBarrier endBarrier;
	private long processStart;

	/*
	 * Start the application.
	 */
	public static void main(String[] args) throws InterruptedException, BrokenBarrierException {
		new ImprovedWorkerDemo();
	}

	/*
	 * The main class that sets up the barriers, queue, buffer, workers and runs
	 * the main logic.
	 */
	public ImprovedWorkerDemo() throws InterruptedException, BrokenBarrierException {

		// All workers drain from this queue, and execute the jobs.
		queue = new ArrayBlockingQueue<Job>(NUMBER_OF_JOBS);

		// We wish to populate this buffer with a number of jobs, and then add
		// it all in one atomic call to the queue.
		Collection<Job> jobs = new ArrayList<Job>(NUMBER_OF_JOBS);
		for (int i = 0; NUMBER_OF_JOBS > i; i++) {
			jobs.add(new Job());
		}

		// For comparison, let's see how fast it is on a single thread.
		long start = System.nanoTime();
		for (Job job : jobs) {
			job.execute();
		}
		System.out.printf("1 thread takes %f ms.\n", (float) (System.nanoTime() - start) / 1000000f);

		// Barrier used to sync workers and main-thread.
		startBarrier = new CyclicBarrier(WORKERS + 1);
		endBarrier = new CyclicBarrier(WORKERS + 1);

		// Create and start the threads. Note, they will not start executing
		// until they get unblocked by the goBarrier.
		Thread[] workThreads = new Thread[WORKERS];
		for (int i = 0; WORKERS > i; i++) {
			workThreads[i] = new Thread(new Worker());
			workThreads[i].start();
		}

		// Run game loop.
		long lastTime = System.currentTimeMillis();
		long acc = 0;
		int fps = 0;
		int frames = 0;
		while (true) {
			long now = System.currentTimeMillis();
			acc += now - lastTime;
			frames++;
			if(acc >= 1000) {
				fps = frames;
				frames = 0;
				acc -= 1000;
				System.out.println("FPS: " + fps);
			}
			
			lastTime = now;
			processJobs(jobs);
		}
	}

	private void processJobs(Collection<Job> jobs) throws InterruptedException, BrokenBarrierException {
		processStart = System.nanoTime();

		//System.out.printf("[%f] Started.\n", (float) (System.nanoTime() - processStart) / 1000000f);
		//System.out.printf("[%f] Now adding the jobs to the queue.\n", (float) (System.nanoTime() - processStart) / 1000000f);
		long addingStart = System.nanoTime();
		queue.addAll(jobs); // TODO we need faster solution for this!
		//System.out.printf("[%f] Adding to the queue took %f ms.\n", (float) (System.nanoTime() - processStart) / 1000000f,
		//		(float) (System.nanoTime() - addingStart) / 1000000f);

		startBarrier.await(); // Signal workers to start working. Unblocks all.
		//System.out.printf("[%f] Workers have started, all unblocked.\n", (float) (System.nanoTime() - processStart) / 1000000f);
		//System.out.printf("[%f] Resetting startBarrier.\n", (float) (System.nanoTime() - processStart) / 1000000f);
		startBarrier.reset();
		//System.out.printf("[%f] Resetting startBarrier is done.\n", (float) (System.nanoTime() - processStart) / 1000000f);

		//System.out.printf("[%f] Waiting for workers to end.\n", (float) (System.nanoTime() - processStart) / 1000000f);
		endBarrier.await(); // Wait for all workers to end.
		//System.out.printf("[%f] Workers have ended.\n", (float) (System.nanoTime() - processStart) / 1000000f);
		//System.out.printf("[%f] Resetting endBarrier.\n", (float) (System.nanoTime() - processStart) / 1000000f);
		endBarrier.reset();
		//System.out.printf("[%f] Resetting endBarrier is done.\n", (float) (System.nanoTime() - processStart) / 1000000f);

		//System.out.printf("[%f] Ended. Took total %f ms.\n\n", (float) (System.nanoTime() - processStart) / 1000000f, (float) (System.nanoTime() - processStart) / 1000000f);
	}

	/*
	 * A dummy Job class that loops a number of times to simulate CPU load.
	 */
	private class Job {
		public void execute() {
			int a = 0;
			for (int i = 0; 1000 > i; i++) {
				a += i;
			}
		}
	}

	/*
	 * Worker class that drains from the queue and runs the jobs, while being in
	 * sync with other threads and primary thread.
	 */
	private class Worker implements Runnable {
		private static final int DRAINCOUNT = 100;
		private final Collection<Job> drainer;

		public Worker() {
			this.drainer = new ArrayList<Job>(DRAINCOUNT);
		}

		@Override
		public void run() {
			while (true) {
				try {
					startBarrier.await(); // Worker is ready.
				} catch (InterruptedException e) {
					e.printStackTrace();
				} catch (BrokenBarrierException e) {
					e.printStackTrace();
				}

				long start = System.nanoTime();
				while (queue.drainTo(drainer, DRAINCOUNT) > 0) {
					for (Job job : drainer) {
						job.execute();
					}
					drainer.clear();
				}
				//System.out.printf("[%f] Done in %f ms.\n", (float) (System.nanoTime() - processStart) / 1000000f, (float) (System.nanoTime() - start) / 1000000f);

				try {
					endBarrier.await(); // Worker is done.
				} catch (InterruptedException e) {
					e.printStackTrace();
				} catch (BrokenBarrierException e) {
					e.printStackTrace();
				}

			}
		}

	}

}
